package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/24 0024.
  * Kaggle2014，从facebook朋友关系中划分社交圈子。
  * 参见电子工业出版社的2017年4月第一版《Spark GraphX in Action》5.4.1
  */
object TestConnectedComponents {
  //Scala小贴士：可以在字符串后面追加一个.r来创建一个正则表达式。在Java中写正则表达式通常比较繁琐，因为反斜杠需要通过另一个反斜杠来转义。
  //而在Scala里提供了用三个双引号表示的原生字符串语法。同事Scala中的正则也提供了抽取器方法，它允许指定一个变量，正则匹配后的结果会赋值
  //给这个变量。代码val Pattern(num)=s表示用正则匹配到字符串s后赋值给变量num。
  //从<path>/<userId>.egonet格式的文件路径中解析出用户ID
  def extract(s: String)={
    val Pattern = """^.*?(\d+).egonet""".r
    val Pattern(num) = s
    num
  }

  //处理egonet文件的每行数据，返回元组形式的边数据。
  def get_edges_from_line(line: String):Array[(Long, Long)]={
    val ary = line.split(":")
    val srcId = ary(0).toInt
    val dstIds = ary(1).split(" ")//就如Java里的String.split()一样这里会根据分隔符把字符串切分为一个数组。
    val edges = for{//Scala中的for循环
      dstId <- dstIds
      if (dstId != "")
    }yield{
      (srcId.toLong,dstId.toLong)
    }
    //一个细节：如果用户与其他任何人没有连接，则生成一个自连接,
    //所以顶点会被包含在Graph.fromEdgeTuples构建的图里面。
    if(edges.size > 0) edges else Array((srcId, srcId))
  }

  //从文件内容中构建边元组
  def make_edges(contents: String)={
    val lines = contents.split("\n")
    val unflat = for{
      line <- lines
    }yield{
      get_edges_from_line(line)
    }
    //要传递给Graph.fromEdgeTuples函数一个元组类型的数组。
    //但现在是一个二维数组，这就需要调用flatten()函数来将其扁平化为一维数组。
    val flat = unflat.flatten
    flat//返回值是array of tuple类型。
  }





  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)

    //从Edge元组构建一个图对象，执行connectedComponents函数，返回String结果。
    def get_circles(flat: Array[(Long,Long)])={
      val edges = sc.makeRDD(flat)
      val g = Graph.fromEdgeTuples(edges,1)
      val cc = g.connectedComponents()
      cc.vertices.map(x => (x._2, Array(x._1))).
        reduceByKey( (a,b) => a++b).
        values.map(_.mkString(" ")).collect.mkString(";")
    }


    //wholeTextFiles函数可以读取一个包含多个文件的目录为一个RDD(PairRDD)对象。
    //当要读取的每个文件都是一个key-value时，key是文件路径，value是文件内容。
    //比如key是data/egonets/0.egonet，而值是1:146 189 229 201....752: 146 191....
    val egonets = sc.wholeTextFiles("data/egonets")
//    println(sc.asInstanceOf[AnyRef].getClass)
    val egonet_numbers = egonets.map(x => extract(x._1)).collect
    val egonet_edges = egonets.map(x => make_edges(x._2))//map内返回的是array of tuple，所以egonet_edges是list of array of tuple，是二维数组。
    val egonet_circles = egonet_edges.map(x => get_circles(x))
    println(egonet_circles.asInstanceOf[AnyRef].getClass)

  }
}
